package cn.lili.controller.socket;

import cn.hutool.json.JSONUtil;
import cn.lili.common.enums.ResultCode;
import cn.lili.common.exception.ServiceException;
import cn.lili.common.security.AuthUser;
import cn.lili.common.security.context.UserContext;
import cn.lili.common.security.enums.SceneEnums;
import cn.lili.modules.im.entity.dos.ImMessage;
import cn.lili.modules.im.entity.dos.ImTalk;
import cn.lili.modules.im.entity.enums.MessageResultType;
import cn.lili.modules.im.entity.vo.MessageOperation;
import cn.lili.modules.im.entity.vo.MessageVO;
import cn.lili.modules.im.service.ImMessageService;
import cn.lili.modules.im.service.ImTalkService;
import com.alibaba.druid.util.StringUtils;
import com.alibaba.fastjson.JSON;
import com.baomidou.mybatisplus.core.conditions.update.LambdaUpdateWrapper;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;
import org.springframework.web.socket.TextMessage;
import org.springframework.web.socket.WebSocketSession;

import java.io.IOException;
import java.util.Map;
import java.util.Optional;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CopyOnWriteArraySet;
import java.util.concurrent.atomic.AtomicInteger;

@Slf4j
@Component
@RequiredArgsConstructor
public class WebSocketImpl implements WebSocket {
    /**
     * 在线人数
     * PS 注意，只能单节点，如果多节点部署需要自行寻找方案
     */
    private static final ConcurrentHashMap<String, WebSocketSession> sessionPools = new ConcurrentHashMap<>();
    private static final String ACCESS_TOKEN = "accessToken";
    /**
     * 在线连接数（线程安全）
     */
    private final AtomicInteger connectionCount = new AtomicInteger(0);
    /**
     * 线程安全的无序集合（存储会话）
     */
    private final CopyOnWriteArraySet<WebSocketSession> sessions = new CopyOnWriteArraySet<>();
    /**
     * 消息服务
     */
    private final ImMessageService imMessageService;
    private final ImTalkService imTalkService;

    @Override
    public void handleOpen(WebSocketSession session) {
        sessions.add(session);
        int count = connectionCount.incrementAndGet();
        log.info("a new connection opened，current online count：{}", count);

        AuthUser authUser = this.getAuthUser(session);

        String sessionId = SceneEnums.STORE.equals(authUser.getScene()) ? authUser.getExtendId() : authUser.getId();
        //如果已有会话，则进行下线提醒。
        if (sessionPools.containsKey(sessionId)) {
            log.info("用户重复登陆，旧用户下线");
            try (WebSocketSession oldSession = sessionPools.get(sessionId)) {
                sendMessage(oldSession, MessageVO.builder().messageResultType(MessageResultType.OFFLINE).result("用户异地登陆").build());
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
        sessionPools.put(sessionId, session);
    }

    @Override
    public void handleClose(WebSocketSession session) {
        sessions.remove(session);
        int count = connectionCount.decrementAndGet();
        log.info("a new connection closed，current online count：{}", count);
        AuthUser authUser = this.getAuthUser(session);
        String sessionId = SceneEnums.STORE.equals(authUser.getScene()) ? authUser.getExtendId() : authUser.getId();
        log.info("用户断开断开连接:{}", JSONUtil.toJsonStr(authUser));
        sessionPools.remove(sessionId);
    }

    @Override
    public void handleMessage(WebSocketSession session, String message) {
        log.info("received a message：{}", message);
        this.operation(session, JSON.parseObject(message, MessageOperation.class));
    }

    @Override
    public void sendMessage(WebSocketSession session, MessageVO message) throws IOException {
        this.sendMessage(session, new TextMessage(JSON.toJSONString(message, true)));
    }

    @Override
    public void sendMessage(String accessToken, TextMessage message) throws IOException {
        Optional<WebSocketSession> userSession = sessions.stream().filter(session -> {
            if (!session.isOpen()) {
                return false;
            }
            Map<String, Object> attributes = session.getAttributes();
            if (!attributes.containsKey(ACCESS_TOKEN)) {
                return false;
            }
            String token = (String) attributes.get(ACCESS_TOKEN);
            return token.equals(accessToken);
        }).findFirst();
        if (userSession.isPresent()) {
            userSession.get().sendMessage(message);
        }
    }

    @Override
    public void sendMessage(String accessToken, String message) throws IOException {
        this.sendMessage(accessToken, new TextMessage(message));
    }

    @Override
    public void sendMessage(WebSocketSession session, TextMessage message) throws IOException {
        try {
            session.sendMessage(message);
        } catch (Exception e) {
            log.info("发送信息对象不在线,信息内容为 {}", message.getPayload());
        }
    }

    /**
     * 发送消息
     *
     * @param sessionId sessionId
     * @param message   消息对象
     */
    private void sendMessage(String sessionId, MessageVO message) {
        try {
            WebSocketSession session = sessionPools.get(sessionId);
            sendMessage(session, message);
        } catch (Exception e) {
            log.error("发送消息失败", e);
        }
    }

    @Override
    public void broadCast(String message) throws IOException {
        for (WebSocketSession session : sessions) {
            if (!session.isOpen()) {
                continue;
            }
            this.sendMessage(session, new TextMessage(message));
        }
    }

    @Override
    public void broadCast(TextMessage message) throws IOException {
        for (WebSocketSession session : sessions) {
            if (!session.isOpen()) {
                continue;
            }
            session.sendMessage(message);
        }
    }

    @Override
    public void handleError(WebSocketSession session, Throwable error) {
        log.error("websocket error：{}，session id：{}", error.getMessage(), session.getId());
        log.error("", error);
    }

    @Override
    public Set<WebSocketSession> getSessions() {
        return sessions;
    }

    @Override
    public int getConnectionCount() {
        return connectionCount.get();
    }


    /**
     * IM操作
     *
     * @param session          session
     * @param messageOperation 操作
     */
    private void operation(WebSocketSession session, MessageOperation messageOperation) {

        Object accessToken = session.getAttributes().get(ACCESS_TOKEN);
        if (accessToken == null) {
            log.error("用户未登录，无法建立连接");
            throw new ServiceException(ResultCode.USER_NOT_LOGIN);
        }
        AuthUser authUser = UserContext.getAuthUser(session.getAttributes().get(ACCESS_TOKEN).toString());
        if (authUser == null) {
            throw new ServiceException(ResultCode.USER_NOT_LOGIN);
        }
        switch (messageOperation.getOperationType()) {
            case PING:
                break;
            case MESSAGE:
                //保存消息
                ImMessage imMessage = new ImMessage(messageOperation);
                imMessageService.save(imMessage);
                //修改最后消息信息
                imTalkService.update(new LambdaUpdateWrapper<ImTalk>().eq(ImTalk::getId, messageOperation.getTalkId()).set(ImTalk::getLastTalkMessage, messageOperation.getContext())
                        .set(ImTalk::getLastTalkTime, imMessage.getCreateTime())
                        .set(ImTalk::getLastMessageType, imMessage.getMessageType()));
                //发送消息
                sendMessage(messageOperation.getTo(), new MessageVO(MessageResultType.MESSAGE, imMessage));
                break;
            case READ:
                if (!StringUtils.isEmpty(messageOperation.getContext())) {
                    imMessageService.read(messageOperation.getTalkId(), accessToken.toString());
                }
                break;
            case UNREAD:
                sendMessage(authUser.getId(), new MessageVO(MessageResultType.UN_READ, imMessageService.unReadMessages(accessToken.toString())));
                break;
            case HISTORY:
                sendMessage(authUser.getId(), new MessageVO(MessageResultType.HISTORY, imMessageService.historyMessage(accessToken.toString(), messageOperation.getTo())));
                break;
            default:
                break;
        }
    }

    private AuthUser getAuthUser(WebSocketSession session) {
        if (session.getAttributes().get(ACCESS_TOKEN) == null) {
            log.error("用户未登录，无法建立连接");
            throw new ServiceException(ResultCode.USER_NOT_LOGIN);
        }
        AuthUser authUser = UserContext.getAuthUser(session.getAttributes().get(ACCESS_TOKEN).toString());
        if (authUser == null) {
            throw new ServiceException(ResultCode.USER_NOT_LOGIN);
        }
        return authUser;
    }
}
